package org.example.sink;

import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.example.aggres.Dog;

import java.util.concurrent.TimeUnit;

/**
 * rich function
 */
public class DataSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Dog> source = env.fromElements(
                new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1),
                new Dog("bb", "b", 5), new Dog("bb", "b", 5), new Dog("bb", "b", 5),
                new Dog("cc", "b", 6), new Dog("cc", "b", 6),
                new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8),
                new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4),
                new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4)

        );
        env.setParallelism(1);
        StreamingFileSink<String> fileSink = StreamingFileSink.<String>forRowFormat(
                        new Path("log1"),
                        new SimpleStringEncoder<>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withMaxPartSize(1024 * 1024 * 1024)
                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
                                .build()
                )
                .build();
        source.map(data -> data.toString()).addSink(fileSink);

        env.execute();
    }
}


// batchinsert